<!DOCTYPE html>
<html lang="en-US">
  <head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width,initial-scale=1">
    <title>Kafka 工作流程和存储机制分析 | JavaKeeper</title>
    <meta name="generator" content="VuePress 1.5.4">
    <link rel="icon" href="/icon.svg">
    <script>
        var _hmt = _hmt || [];
        (function() {
            var hm = document.createElement("script");
            hm.src = "https://hm.baidu.com/hm.js?a949a9b30eb86ac0159e735ff8670c03";
            var s = document.getElementsByTagName("script")[0];
            s.parentNode.insertBefore(hm, s);
            // 引入谷歌,不需要可删除这段
            var hm1 = document.createElement("script");
            hm1.src = "https://www.googletagmanager.com/gtag/js?id=UA-169923503-1";
            var s1 = document.getElementsByTagName("script")[0]; 
            s1.parentNode.insertBefore(hm1, s1);
        })();
        // 谷歌加载,不需要可删除
        window.dataLayer = window.dataLayer || [];
        function gtag(){dataLayer.push(arguments);}
        gtag('js', new Date());
        gtag('config', 'UA-169923503-1');
    </script>
    <meta name="description" content="">
    <meta name="keywords" content="JavaKeeper,Java,Java开发,算法,blog">
    <link rel="preload" href="/assets/css/0.styles.91f57736.css" as="style"><link rel="preload" href="/assets/js/app.447d4224.js" as="script"><link rel="preload" href="/assets/js/3.9d76740c.js" as="script"><link rel="preload" href="/assets/js/1.c4fd7d2e.js" as="script"><link rel="preload" href="/assets/js/155.c68bf602.js" as="script"><link rel="prefetch" href="/assets/js/10.8cf3be2c.js"><link rel="prefetch" href="/assets/js/100.74f35ab8.js"><link rel="prefetch" href="/assets/js/101.7a062346.js"><link rel="prefetch" href="/assets/js/102.c9485235.js"><link rel="prefetch" href="/assets/js/103.d88a3805.js"><link rel="prefetch" href="/assets/js/104.6e034144.js"><link rel="prefetch" href="/assets/js/105.d22f7450.js"><link rel="prefetch" href="/assets/js/106.a6cb54b0.js"><link rel="prefetch" href="/assets/js/107.7b65e72b.js"><link rel="prefetch" href="/assets/js/108.eb5804bb.js"><link rel="prefetch" href="/assets/js/109.05f775e5.js"><link rel="prefetch" href="/assets/js/11.c54ae13c.js"><link rel="prefetch" href="/assets/js/110.51d3d641.js"><link rel="prefetch" href="/assets/js/111.022b64a7.js"><link rel="prefetch" href="/assets/js/112.da8afd52.js"><link rel="prefetch" href="/assets/js/113.05a17b18.js"><link rel="prefetch" href="/assets/js/114.8960d913.js"><link rel="prefetch" href="/assets/js/115.67919f68.js"><link rel="prefetch" href="/assets/js/116.62b0cd71.js"><link rel="prefetch" href="/assets/js/117.ebac3eff.js"><link rel="prefetch" href="/assets/js/118.ecd629bd.js"><link rel="prefetch" href="/assets/js/119.a09a0897.js"><link rel="prefetch" href="/assets/js/12.60aa3b24.js"><link rel="prefetch" href="/assets/js/120.bf639d3d.js"><link rel="prefetch" href="/assets/js/121.b89d0c8e.js"><link rel="prefetch" href="/assets/js/122.1a75ff83.js"><link rel="prefetch" href="/assets/js/123.d2127132.js"><link rel="prefetch" href="/assets/js/124.2caff9e0.js"><link rel="prefetch" href="/assets/js/125.9b9f966a.js"><link rel="prefetch" href="/assets/js/126.58cdfb3d.js"><link rel="prefetch" href="/assets/js/127.8ef09c53.js"><link rel="prefetch" href="/assets/js/128.efdc2ae4.js"><link rel="prefetch" href="/assets/js/129.e35cbc57.js"><link rel="prefetch" href="/assets/js/13.125c13a0.js"><link rel="prefetch" href="/assets/js/130.f01a55e3.js"><link rel="prefetch" href="/assets/js/131.65205f4a.js"><link rel="prefetch" href="/assets/js/132.f42c5a0a.js"><link rel="prefetch" href="/assets/js/133.9ba468b3.js"><link rel="prefetch" href="/assets/js/134.7b597ba9.js"><link rel="prefetch" href="/assets/js/135.fb828b9a.js"><link rel="prefetch" href="/assets/js/136.3887532f.js"><link rel="prefetch" href="/assets/js/137.549bae01.js"><link rel="prefetch" href="/assets/js/138.db8d423d.js"><link rel="prefetch" href="/assets/js/139.dbaf2267.js"><link rel="prefetch" href="/assets/js/14.bd1d0b0d.js"><link rel="prefetch" href="/assets/js/140.6cb65fdc.js"><link rel="prefetch" href="/assets/js/141.9bd6cc4b.js"><link rel="prefetch" href="/assets/js/142.552db5ed.js"><link rel="prefetch" href="/assets/js/143.2c9f2bf4.js"><link rel="prefetch" href="/assets/js/144.fba98a15.js"><link rel="prefetch" href="/assets/js/145.c42f3a21.js"><link rel="prefetch" href="/assets/js/146.596d4d33.js"><link rel="prefetch" href="/assets/js/147.c48ae5c1.js"><link rel="prefetch" href="/assets/js/148.71064871.js"><link rel="prefetch" href="/assets/js/149.16582d21.js"><link rel="prefetch" href="/assets/js/15.f247873b.js"><link rel="prefetch" href="/assets/js/150.ead09aca.js"><link rel="prefetch" href="/assets/js/151.971fdf4b.js"><link rel="prefetch" href="/assets/js/152.369c9362.js"><link rel="prefetch" href="/assets/js/153.371edd15.js"><link rel="prefetch" href="/assets/js/154.e090b491.js"><link rel="prefetch" href="/assets/js/156.304aea8d.js"><link rel="prefetch" href="/assets/js/157.83beef7f.js"><link rel="prefetch" href="/assets/js/158.bb1794b0.js"><link rel="prefetch" href="/assets/js/159.2d54e792.js"><link rel="prefetch" href="/assets/js/16.04336c71.js"><link rel="prefetch" href="/assets/js/160.99d56586.js"><link rel="prefetch" href="/assets/js/161.edf660aa.js"><link rel="prefetch" href="/assets/js/162.0b84606e.js"><link rel="prefetch" href="/assets/js/163.b59e0d60.js"><link rel="prefetch" href="/assets/js/164.d9eb8228.js"><link rel="prefetch" href="/assets/js/165.ca624c79.js"><link rel="prefetch" href="/assets/js/166.025b2ba1.js"><link rel="prefetch" href="/assets/js/167.abc982cc.js"><link rel="prefetch" href="/assets/js/168.27ca13dc.js"><link rel="prefetch" href="/assets/js/169.41e753a2.js"><link rel="prefetch" href="/assets/js/17.43b3c1c8.js"><link rel="prefetch" href="/assets/js/170.626319e1.js"><link rel="prefetch" href="/assets/js/171.a221dddf.js"><link rel="prefetch" href="/assets/js/172.464b2361.js"><link rel="prefetch" href="/assets/js/173.96a3afee.js"><link rel="prefetch" href="/assets/js/174.116607c2.js"><link rel="prefetch" href="/assets/js/175.ea3e8659.js"><link rel="prefetch" href="/assets/js/176.7d7b8afc.js"><link rel="prefetch" href="/assets/js/177.a6e00aa0.js"><link rel="prefetch" href="/assets/js/178.1f93afaf.js"><link rel="prefetch" href="/assets/js/179.3aa00dcd.js"><link rel="prefetch" href="/assets/js/18.d81b44d5.js"><link rel="prefetch" href="/assets/js/180.f8b2b75a.js"><link rel="prefetch" href="/assets/js/181.8e11258a.js"><link rel="prefetch" href="/assets/js/182.22243941.js"><link rel="prefetch" href="/assets/js/183.d051fdf6.js"><link rel="prefetch" href="/assets/js/184.a994075e.js"><link rel="prefetch" href="/assets/js/185.776c7e16.js"><link rel="prefetch" href="/assets/js/186.f1887955.js"><link rel="prefetch" href="/assets/js/187.da0d3626.js"><link rel="prefetch" href="/assets/js/188.8dfc358f.js"><link rel="prefetch" href="/assets/js/189.dcac5a59.js"><link rel="prefetch" href="/assets/js/19.1b3d66e1.js"><link rel="prefetch" href="/assets/js/190.c7e413d0.js"><link rel="prefetch" href="/assets/js/191.d9806121.js"><link rel="prefetch" href="/assets/js/192.869791da.js"><link rel="prefetch" href="/assets/js/193.2d74c4c8.js"><link rel="prefetch" href="/assets/js/194.c73a1909.js"><link rel="prefetch" href="/assets/js/195.e8c74834.js"><link rel="prefetch" href="/assets/js/20.bd5949ec.js"><link rel="prefetch" href="/assets/js/21.3fcf98cf.js"><link rel="prefetch" href="/assets/js/22.2fa1e2e8.js"><link rel="prefetch" href="/assets/js/23.1ae64bb4.js"><link rel="prefetch" href="/assets/js/24.7bdf7387.js"><link rel="prefetch" href="/assets/js/25.392c436e.js"><link rel="prefetch" href="/assets/js/26.58acbd4b.js"><link rel="prefetch" href="/assets/js/27.c725bdd5.js"><link rel="prefetch" href="/assets/js/28.6c9bda1e.js"><link rel="prefetch" href="/assets/js/29.e656b537.js"><link rel="prefetch" href="/assets/js/30.2c326fc7.js"><link rel="prefetch" href="/assets/js/31.e6c9fa30.js"><link rel="prefetch" href="/assets/js/32.c9c88437.js"><link rel="prefetch" href="/assets/js/33.0c53373c.js"><link rel="prefetch" href="/assets/js/34.9821e543.js"><link rel="prefetch" href="/assets/js/35.de8253eb.js"><link rel="prefetch" href="/assets/js/36.d182f929.js"><link rel="prefetch" href="/assets/js/37.9fa79014.js"><link rel="prefetch" href="/assets/js/38.9bebff76.js"><link rel="prefetch" href="/assets/js/39.19a3a2d4.js"><link rel="prefetch" href="/assets/js/4.564edb9d.js"><link rel="prefetch" href="/assets/js/40.cca6955f.js"><link rel="prefetch" href="/assets/js/41.854cd09a.js"><link rel="prefetch" href="/assets/js/42.ca7b612f.js"><link rel="prefetch" href="/assets/js/43.87027d58.js"><link rel="prefetch" href="/assets/js/44.8c2b4f4b.js"><link rel="prefetch" href="/assets/js/45.dffb4e08.js"><link rel="prefetch" href="/assets/js/46.f58049a5.js"><link rel="prefetch" href="/assets/js/47.6854070c.js"><link rel="prefetch" href="/assets/js/48.6cd9fa3d.js"><link rel="prefetch" href="/assets/js/49.e8861afa.js"><link rel="prefetch" href="/assets/js/5.5c31d62f.js"><link rel="prefetch" href="/assets/js/50.703bffab.js"><link rel="prefetch" href="/assets/js/51.6655c373.js"><link rel="prefetch" href="/assets/js/52.deb2eb09.js"><link rel="prefetch" href="/assets/js/53.6e0ed77d.js"><link rel="prefetch" href="/assets/js/54.b05c58ad.js"><link rel="prefetch" href="/assets/js/55.49c8164e.js"><link rel="prefetch" href="/assets/js/56.a5574e6b.js"><link rel="prefetch" href="/assets/js/57.58cb0de4.js"><link rel="prefetch" href="/assets/js/58.52345112.js"><link rel="prefetch" href="/assets/js/59.663ce78d.js"><link rel="prefetch" href="/assets/js/6.a9df34ee.js"><link rel="prefetch" href="/assets/js/60.f06adde2.js"><link rel="prefetch" href="/assets/js/61.170255a1.js"><link rel="prefetch" href="/assets/js/62.9d120050.js"><link rel="prefetch" href="/assets/js/63.70cced6b.js"><link rel="prefetch" href="/assets/js/64.577f3548.js"><link rel="prefetch" href="/assets/js/65.c037b29d.js"><link rel="prefetch" href="/assets/js/66.7dd1045f.js"><link rel="prefetch" href="/assets/js/67.d3aa4d6c.js"><link rel="prefetch" href="/assets/js/68.526dbb61.js"><link rel="prefetch" href="/assets/js/69.58269266.js"><link rel="prefetch" href="/assets/js/7.6609d4d6.js"><link rel="prefetch" href="/assets/js/70.64108f1b.js"><link rel="prefetch" href="/assets/js/71.1e95e0a6.js"><link rel="prefetch" href="/assets/js/72.42e7ec94.js"><link rel="prefetch" href="/assets/js/73.dad4e1c5.js"><link rel="prefetch" href="/assets/js/74.28ea286a.js"><link rel="prefetch" href="/assets/js/75.dd6d4c6f.js"><link rel="prefetch" href="/assets/js/76.ca6539df.js"><link rel="prefetch" href="/assets/js/77.feb13b0e.js"><link rel="prefetch" href="/assets/js/78.321e90e6.js"><link rel="prefetch" href="/assets/js/79.68eb8fcf.js"><link rel="prefetch" href="/assets/js/8.396d51fd.js"><link rel="prefetch" href="/assets/js/80.4edb5321.js"><link rel="prefetch" href="/assets/js/81.735d7e57.js"><link rel="prefetch" href="/assets/js/82.fa120bdf.js"><link rel="prefetch" href="/assets/js/83.bf755f94.js"><link rel="prefetch" href="/assets/js/84.9b32070c.js"><link rel="prefetch" href="/assets/js/85.592aca7c.js"><link rel="prefetch" href="/assets/js/86.4dcd9e73.js"><link rel="prefetch" href="/assets/js/87.a9e546aa.js"><link rel="prefetch" href="/assets/js/88.2a423212.js"><link rel="prefetch" href="/assets/js/89.5f455115.js"><link rel="prefetch" href="/assets/js/9.adb074c6.js"><link rel="prefetch" href="/assets/js/90.5202da0a.js"><link rel="prefetch" href="/assets/js/91.02cee99d.js"><link rel="prefetch" href="/assets/js/92.f16bad0b.js"><link rel="prefetch" href="/assets/js/93.f933634f.js"><link rel="prefetch" href="/assets/js/94.8e7b1d65.js"><link rel="prefetch" href="/assets/js/95.ee0e4a0a.js"><link rel="prefetch" href="/assets/js/96.e21d78c2.js"><link rel="prefetch" href="/assets/js/97.c87e514e.js"><link rel="prefetch" href="/assets/js/98.d123ac92.js"><link rel="prefetch" href="/assets/js/99.92d1b416.js">
    <link rel="stylesheet" href="/assets/css/0.styles.91f57736.css">
  </head>
  <body>
    <div id="app" data-server-rendered="true"><div class="theme-container no-sidebar" data-v-3ba18f14><div data-v-3ba18f14><div id="loader-wrapper" class="loading-wrapper" data-v-041fef5b data-v-3ba18f14 data-v-3ba18f14><div class="loader-main" data-v-041fef5b><div data-v-041fef5b></div><div data-v-041fef5b></div><div data-v-041fef5b></div><div data-v-041fef5b></div></div> <!----> <!----></div> <div class="password-shadow password-wrapper-out" style="display:none;" data-v-68139a52 data-v-3ba18f14 data-v-3ba18f14><h3 class="title" style="display:none;" data-v-68139a52 data-v-68139a52>JavaKeeper</h3> <!----> <label id="box" class="inputBox" style="display:none;" data-v-68139a52 data-v-68139a52><input type="password" value="" data-v-68139a52> <span data-v-68139a52>Konck! Knock!</span> <button data-v-68139a52>OK</button></label> <div class="footer" style="display:none;" data-v-68139a52 data-v-68139a52><span data-v-68139a52><i class="iconfont reco-theme" data-v-68139a52></i> <a target="blank" href="https://vuepress-theme-reco.recoluan.com" data-v-68139a52>vuePress-theme-reco</a></span> <span data-v-68139a52><i class="iconfont reco-copyright" data-v-68139a52></i> <a data-v-68139a52><span data-v-68139a52>海星</span>
            
          <!---->
          2020
        </a></span></div></div> <div class="hide" data-v-3ba18f14><header class="navbar" data-v-3ba18f14><div class="sidebar-button"><svg xmlns="http://www.w3.org/2000/svg" aria-hidden="true" role="img" viewBox="0 0 448 512" class="icon"><path fill="currentColor" d="M436 124H12c-6.627 0-12-5.373-12-12V80c0-6.627 5.373-12 12-12h424c6.627 0 12 5.373 12 12v32c0 6.627-5.373 12-12 12zm0 160H12c-6.627 0-12-5.373-12-12v-32c0-6.627 5.373-12 12-12h424c6.627 0 12 5.373 12 12v32c0 6.627-5.373 12-12 12zm0 160H12c-6.627 0-12-5.373-12-12v-32c0-6.627 5.373-12 12-12h424c6.627 0 12 5.373 12 12v32c0 6.627-5.373 12-12 12z"></path></svg></div> <a href="/" class="home-link router-link-active"><!----> <span class="site-name">JavaKeeper</span></a> <div class="links"><div class="color-picker"><a class="color-button"><i class="iconfont reco-color"></i></a> <div class="color-picker-menu" style="display:none;"><div class="mode-options"><h4 class="title">Choose mode</h4> <ul class="color-mode-options"><li class="dark">dark</li><li class="auto active">auto</li><li class="light">light</li></ul></div></div></div> <div class="search-box"><i class="iconfont reco-search"></i> <input aria-label="Search" autocomplete="off" spellcheck="false" value=""> <!----></div> <nav class="nav-links can-hide"><div class="nav-item"><a href="/java/" class="nav-link"><i class="iconfont undefined"></i>
  Java
</a></div><div class="nav-item"><a href="/data-structure-algorithms/" class="nav-link"><i class="iconfont undefined"></i>
  数据结构与算法
</a></div><div class="nav-item"><a href="/data-store/" class="nav-link"><i class="iconfont undefined"></i>
  数据存储与缓存
</a></div><div class="nav-item"><a href="/interview/" class="nav-link"><i class="iconfont undefined"></i>
  直击面试
</a></div> <a href="https://github.com/Jstarfish/JavaKeeper" target="_blank" rel="noopener noreferrer" class="repo-link"><i class="iconfont reco-github"></i>
    GitHub
    <svg xmlns="http://www.w3.org/2000/svg" aria-hidden="true" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15" class="icon outbound"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path> <polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg></a></nav></div></header> <div class="sidebar-mask" data-v-3ba18f14></div> <aside class="sidebar" data-v-3ba18f14><div class="personal-info-wrapper" data-v-5f6acefd data-v-3ba18f14><!----> <h3 class="name" data-v-5f6acefd>
    海星
  </h3> <div class="num" data-v-5f6acefd><div data-v-5f6acefd><h3 data-v-5f6acefd>0</h3> <h6 data-v-5f6acefd>Article</h6></div> <div data-v-5f6acefd><h3 data-v-5f6acefd>0</h3> <h6 data-v-5f6acefd>Tag</h6></div></div> <hr data-v-5f6acefd></div> <nav class="nav-links"><div class="nav-item"><a href="/java/" class="nav-link"><i class="iconfont undefined"></i>
  Java
</a></div><div class="nav-item"><a href="/data-structure-algorithms/" class="nav-link"><i class="iconfont undefined"></i>
  数据结构与算法
</a></div><div class="nav-item"><a href="/data-store/" class="nav-link"><i class="iconfont undefined"></i>
  数据存储与缓存
</a></div><div class="nav-item"><a href="/interview/" class="nav-link"><i class="iconfont undefined"></i>
  直击面试
</a></div> <a href="https://github.com/Jstarfish/JavaKeeper" target="_blank" rel="noopener noreferrer" class="repo-link"><i class="iconfont reco-github"></i>
    GitHub
    <svg xmlns="http://www.w3.org/2000/svg" aria-hidden="true" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15" class="icon outbound"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path> <polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg></a></nav> <!----> </aside> <div class="password-shadow password-wrapper-in" style="display:none;" data-v-68139a52 data-v-3ba18f14><h3 class="title" style="display:none;" data-v-68139a52 data-v-68139a52></h3> <!----> <label id="box" class="inputBox" style="display:none;" data-v-68139a52 data-v-68139a52><input type="password" value="" data-v-68139a52> <span data-v-68139a52>Konck! Knock!</span> <button data-v-68139a52>OK</button></label> <div class="footer" style="display:none;" data-v-68139a52 data-v-68139a52><span data-v-68139a52><i class="iconfont reco-theme" data-v-68139a52></i> <a target="blank" href="https://vuepress-theme-reco.recoluan.com" data-v-68139a52>vuePress-theme-reco</a></span> <span data-v-68139a52><i class="iconfont reco-copyright" data-v-68139a52></i> <a data-v-68139a52><span data-v-68139a52>海星</span>
            
          <!---->
          2020
        </a></span></div></div> <div data-v-3ba18f14><main class="page"><div class="page-title" style="display:none;"><h1 class="title">Kafka 工作流程和存储机制分析</h1> <div data-v-5d8dbdb4><i class="iconfont reco-account" data-v-5d8dbdb4><span data-v-5d8dbdb4>海星</span></i> <!----> <!----> <!----></div></div> <div class="theme-reco-content content__default" style="display:none;"><h2 id="kafka-工作流程和存储机制分析"><a href="#kafka-工作流程和存储机制分析" class="header-anchor">#</a> Kafka 工作流程和存储机制分析</h2> <p><img src="https://images.gitbook.cn/e49bc290-cf95-11e8-8388-bd48f25029c6" alt=""></p> <h3 id="一、kafka-文件存储机制"><a href="#一、kafka-文件存储机制" class="header-anchor">#</a> 一、Kafka 文件存储机制</h3> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh45cng275j30ny0bjwfr.jpg" alt=""></p> <h4 id="topic构成"><a href="#topic构成" class="header-anchor">#</a> topic构成</h4> <p><strong>Kafka 中消息是以 topic 进行分类的</strong>，生产者生产消息，消费者消费消息，都是面向 topic 的。</p> <p>在 Kafka 中，一个 topic 可以分为多个 partition，一个 partition 分为多个 <strong>segment</strong>，每个 segment 对应两个文件：.index 和 .log 文件</p> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh3ntjmvt8j31600mzac4.jpg" alt=""></p> <p><strong>topic 是逻辑上的概念，而 patition 是物理上的概念</strong>，每个 patition 对应一个 log 文件，而 log 文件中存储的就是 producer 生产的数据，patition 生产的数据会被不断的添加到 log 文件的末端，且每条数据都有自己的 offset。</p> <p>消费组中的每个消费者，都是实时记录自己消费到哪个 offset，以便出错恢复，从上次的位置继续消费。</p> <blockquote><p>问：Kafka 为什么要将 Topic 进行分区</p> <p>答：负载均衡+水平扩展</p> <p>Topic 只是逻辑概念，面向的是 producer 和 consumer；而 Partition 则是物理概念。可以想象，如果 Topic 不进行分区，而将 Topic 内的消息存储于一个 broker，那么关于该 Topic 的所有读写请求都将由这一个 broker 处理，吞吐量很容易陷入瓶颈，这显然是不符合高吞吐量应用场景的。有了 Partition 概念以后，假设一个 Topic 被分为 10 个 Partitions，Kafka 会根据一定的算法将 10 个 Partition 尽可能均匀的分布到不同的 broker（服务器）上，当 producer 发布消息时，producer 客户端可以采用 <code>random</code>、<code>key-hash</code> 及 <code>轮询</code> 等算法选定目标 partition，若不指定，Kafka 也将根据一定算法将其置于某一分区上。Partiton 机制可以极大的提高吞吐量，并且使得系统具备良好的水平扩展能力。</p> <p>在创建 topic 时可以在 <code>$KAFKA_HOME/config/server.properties</code> 中指定这个 partition 的数量（如下所示），当然可以在 topic 创建之后去修改 partition 的数量。</p> <div class="language-properties extra-class"><pre class="language-properties"><code><span class="token comment"># The default number of log partitions per topic. More partitions allow greater</span>
<span class="token comment"># parallelism for consumption, but this will also result in more files across</span>
<span class="token comment"># the brokers.</span>
<span class="token attr-name">num.partitions</span><span class="token punctuation">=</span><span class="token attr-value">3</span>
</code></pre></div><p>在发送一条消息时，可以指定这个消息的 key，producer 根据这个 key 和 partition 机制来判断这个消息发送到哪个partition。partition 机制可以通过指定 producer 的 partition.class 这一参数来指定（即支持自定义），该 class 必须实现 kafka.producer.Partitioner 接口。</p></blockquote> <h4 id="消息存储原理"><a href="#消息存储原理" class="header-anchor">#</a> 消息存储原理</h4> <p>由于生产者生产的消息会不断追加到 log 文件末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采取了<strong>分片</strong>和<strong>索引</strong>机制，将每个 partition 分为多个 segment。每个 segment 对应两个文件——<code>.index文件</code>和 <code>.log文件</code>。这些文件位于一个文件夹下，该文件夹的命名规则为：<code>topic名称+分区序号</code>。</p> <p>如下，我们创建一个只有一个分区一个副本的 topic</p> <div class="language- extra-class"><pre class="language-text"><code>&gt; bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic starfish
</code></pre></div><p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh3lsiywl1j31p804qdjk.jpg" alt=""></p> <p>然后可以在 kafka-logs 目录（server.properties 默认配置）下看到会有个名为 starfish-0 的文件夹。如果，starfish 这个 topic 有三个分区，则其对应的文件夹为 starfish-0，starfish-1，starfish-2。</p> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh3lw95aahj30oy0bm764.jpg" alt=""></p> <p>这些文件的含义如下：</p> <table><thead><tr><th style="text-align:left;">类别</th> <th style="text-align:left;">作用</th></tr></thead> <tbody><tr><td style="text-align:left;">.index</td> <td style="text-align:left;">基于偏移量的索引文件，存放着消息的offset和其对应的物理位置，是**<mark>稀松索引</mark>**</td></tr> <tr><td style="text-align:left;">.timestamp</td> <td style="text-align:left;">时间戳索引文件</td></tr> <tr><td style="text-align:left;">.log</td> <td style="text-align:left;">它是segment文件的数据文件，用于存储实际的消息。该文件是二进制格式的。log文件是存储在 ConcurrentSkipListMap 里的，是一个map结构，key是文件名（offset），value是内容，这样在查找指定偏移量的消息时，用二分查找法就能快速定位到消息所在的数据文件和索引文件</td></tr> <tr><td style="text-align:left;">.snaphot</td> <td style="text-align:left;">快照文件</td></tr> <tr><td style="text-align:left;">leader-epoch-checkpoint</td> <td style="text-align:left;">保存了每一任leader开始写入消息时的offset，会定时更新。 follower被选为leader时会根据这个确定哪些消息可用</td></tr></tbody></table> <p>index 和 log 文件以当前 segment 的第一条消息的 offset 命名。偏移量 offset 是一个 64 位的长整形数，固定是20 位数字，长度未达到，用 0 进行填补，索引文件和日志文件都由此作为文件名命名规则。所以从上图可以看出，我们的偏移量是从 0 开始的，<code>.index</code> 和 <code>.log</code> 文件名称都为 <code>00000000000000000000</code>。</p> <p>接着往 topic 中发送一些消息，并启动消费者消费</p> <div class="language- extra-class"><pre class="language-text"><code>&gt; bin /kafka-console-producer.sh --bootstrap-server localhost:9092 --topic starfish
one
</code></pre></div><div class="language- extra-class"><pre class="language-text"><code>&gt; bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic starfish --from-beginning
one
</code></pre></div><p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh3md0mmslj31y40gcwqe.jpg" alt=""></p> <p>查看 .log 文件下是否有数据 one</p> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh3mujbqwkj30zu082775.jpg" alt=""></p> <p>内容存在一些”乱码“，因为数据是经过序列化压缩的。</p> <p>那么数据文件 .log 大小有限制吗，能保存多久时间？这些我们都可以通过 Kafka 目录下 <code>conf/server.properties</code> 配置文件修改：</p> <div class="language- extra-class"><pre class="language-text"><code># log文件存储时间，单位为小时，这里设置为1周
log.retention.hours=168

# log文件大小的最大值，这里为1g，超过这个值，则会创建新的segment（也就是新的.index和.log文件）
log.segment.bytes=1073741824
</code></pre></div><p>比如，当生产者生产数据量较多，一个 segment 存储不下触发分片时，在日志 topic 目录下你会看到类似如下所示的文件：</p> <div class="language- extra-class"><pre class="language-text"><code>00000000000000000000.index
00000000000000000000.log
00000000000000170410.index
00000000000000170410.log
00000000000000239430.index
00000000000000239430.log
</code></pre></div><ul><li>每个分区是由多个 Segment 组成，当 Kafka 要写数据到一个 partition 时，它会写入到状态为 active 的segment 中。如果该 segment 被写满，则一个新的 segment 将会被新建，然后变成新的“active” segment</li> <li>偏移量：分区中的每一条消息都会被分配的一个连续的id值，该值用于唯一标识分区中的每一条消息</li> <li>每个 Segment 中则保存了真实的消息数据。每个 Segment 对应于一个索引文件与一个日志文件。Segment 文件的生命周期是由 Kafka Server 的配置参数所决定的。比如说，<code>server.properties</code> 文件中的参数项<code>log.retention.hours=168</code> 就表示 7 天后删除老的消息文件</li> <li>[稀松索引]：稀松索引可以加快速度，因为 index 不是为每条消息都存一条索引信息，而是每隔几条数据才存一条 index 信息，这样 index 文件其实很小。kafka在写入日志文件的时候，同时会写索引文件（.index和.timeindex）。默认情况下，有个参数log.index.interval.bytes限定了在日志文件写入多少数据，就要在索引文件写一条索引，默认是4KB，写4kb的数据然后在索引里写一条索引。</li></ul> <p>举个栗子：00000000000000170410 的 “.index” 文件和 “.log” 文件的对应的关系，如下图</p> <p><img src="https://images.gitbook.cn/60eafc10-cc9b-11e8-b452-15eec1b99303" alt=""></p> <blockquote><p>问：为什么不能以 partition 作为存储单位？还要加个 segment？</p> <p>答：如果就以 partition 为最小存储单位，可以想象，当 Kafka producer 不断发送消息，必然会引起 partition 文件的无限扩张，将对消息文件的维护以及已消费的消息的清理带来严重的影响，因此，需以 segment 为单位将 partition 进一步细分。每个 partition（目录）相当于一个巨型文件被平均分配到多个大小相等的 segment（段）数据文件中（每个 segment 文件中消息数量不一定相等）这种特性也方便 old segment 的删除，即方便已被消费的消息的清理，提高磁盘的利用率。每个 partition 只需要支持顺序读写就行，segment 的文件生命周期由服务端配置参数（log.segment.bytes，log.roll.{ms,hours} 等若干参数）决定。</p> <p>问：segment 的工作原理是怎样的？</p> <p>答：segment 文件由两部分组成，分别为 “.index” 文件和 “.log” 文件，分别表示为 segment 索引文件和数据文件。这两个文件的命令规则为：partition 全局的第一个 segment 从 0 开始，后续每个 segment 文件名为上一个 segment 文件最后一条消息的 offset 值，数值大小为 64 位，20 位数字字符长度，没有数字用 0 填充</p></blockquote> <p>下图展示了Kafka查找数据的过程：</p> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh4aushr80j30n60b4aab.jpg" alt=""></p> <p><code>.index文件</code> 存储大量的索引信息，<code>.log文件</code> 存储大量的数据，索引文件中的元数据指向对应数据文件中 message 的物理偏移地址。</p> <p>比如现在要查找偏移量 offset 为 3 的消息，根据 .index 文件命名我们可以知道，offset 为 3 的索引应该从00000000000000000000.index 里查找。根据上图所示，其对应的索引地址为 756-911，所以 Kafka 将读取00000000000000000000.log 756~911区间的数据。</p> <h3 id="二、kafka-生产过程"><a href="#二、kafka-生产过程" class="header-anchor">#</a> 二、Kafka 生产过程</h3> <p>Kafka 生产者用于生产消息。通过前面的内容我们知道，Kafka 的 topic 可以有多个分区，那么生产者如何将这些数据可靠地发送到这些分区？生产者发送数据的不同的分区的依据是什么？针对这两个疑问，这节简单记录下。</p> <h4 id="_3-2-1-写入流程"><a href="#_3-2-1-写入流程" class="header-anchor">#</a> 3.2.1 写入流程</h4> <p>producer 写入消息流程如下：</p> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh45yc0vp8j30zz0gbdik.jpg" alt=""></p> <ol><li>producer 先从 zookeeper 的 &quot;/brokers/.../state&quot;节点找到该 partition 的 leader</li> <li>producer 将消息发送给该 leader</li> <li>leader 将消息写入本地 log</li> <li>followers 从 leader pull 消息，写入本地 log 后向 leader 发送 ACK</li> <li>leader 收到所有 ISR 中的 replication 的 ACK 后，增加 HW（high watermark，最后 commit 的 offset）并向 producer 发送 ACK</li></ol> <h4 id="_2-1-写入方式"><a href="#_2-1-写入方式" class="header-anchor">#</a> 2.1 写入方式</h4> <p>producer 采用推（push） 模式将消息发布到 broker，每条消息都被追加（append） 到分区（patition） 中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障 kafka 吞吐率）。</p> <h4 id="_2-2-分区-partition"><a href="#_2-2-分区-partition" class="header-anchor">#</a> 2.2 分区（Partition）</h4> <p>消息发送时都被发送到一个 topic，其本质就是一个目录，而 topic 是由一些 Partition Logs(分区日志)组成</p> <p><strong>分区的原因：</strong></p> <ol><li><p><strong>方便在集群中扩展</strong>，每个 Partition 可以通过调整以适应它所在的机器，而一个 topic 又可以有多个 Partition 组成，因此整个集群就可以适应任意大小的数据了；</p></li> <li><p><strong>可以提高并发</strong>，因为可以以 Partition 为单位读写了。</p></li></ol> <p><strong>分区的原则：</strong></p> <p>我们需要将 producer 发送的数据封装成一个 ProducerRecord 对象。</p> <div class="language-java extra-class"><pre class="language-java"><code><span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Integer</span> partition<span class="token punctuation">,</span> <span class="token class-name">Long</span> timestamp<span class="token punctuation">,</span> <span class="token class-name">K</span> key<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">,</span> <span class="token class-name">Iterable</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">Header</span><span class="token punctuation">&gt;</span></span> headers<span class="token punctuation">)</span>
<span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Integer</span> partition<span class="token punctuation">,</span> <span class="token class-name">Long</span> timestamp<span class="token punctuation">,</span> <span class="token class-name">K</span> key<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">)</span>
<span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Integer</span> partition<span class="token punctuation">,</span> <span class="token class-name">K</span> key<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">,</span> <span class="token class-name">Iterable</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">Header</span><span class="token punctuation">&gt;</span></span> headers<span class="token punctuation">)</span>
<span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">Integer</span> partition<span class="token punctuation">,</span> <span class="token class-name">K</span> key<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">)</span>
<span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">K</span> key<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">)</span>
<span class="token keyword">public</span> <span class="token class-name">ProducerRecord</span> <span class="token punctuation">(</span><span class="token class-name">String</span> topic<span class="token punctuation">,</span> <span class="token class-name">V</span> value<span class="token punctuation">)</span>
</code></pre></div><ol><li>指明 partition 的情况下，直接将指明的值直接作为 partiton 值；</li> <li>没有指明 partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值；</li> <li>既没有 partition 值又没有 key 值的情况下，第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition 值，也就是常说的 round-robin 算法。</li></ol> <h4 id="_2-3-副本-replication"><a href="#_2-3-副本-replication" class="header-anchor">#</a> 2.3 副本（Replication）</h4> <p>同一个 partition 可能会有多个 replication（ 对应 server.properties 配置中的 <code>default.replication.factor=N</code>）。没有 replication 的情况下，一旦 broker 宕机，其上所有 patition 的数据都不可被消费，同时 producer 也不能再将数据存于其上的 patition。引入 replication 之后，同一个 partition 可能会有多个 replication，而这时需要在这些 replication 之间选出一 个 leader， producer 和 consumer 只与这个 leader 交互，其它 replication 作为 follower 从 leader 中复制数据。</p> <p>为了提高消息的可靠性，Kafka 每个 topic 的 partition 有 N 个副本（replicas），其中 N（大于等于 1）是 topic 的复制因子（replica fator）的个数。Kafka 通过多副本机制实现故障自动转移，当 Kafka 集群中出现 broker 失效时，副本机制可保证服务可用。对于任何一个 partition，它的 N 个 replicas 中，其中一个 replica 为 leader，其他都为 follower，leader 负责处理 partition 的所有读写请求，follower 则负责被动地去复制 leader 上的数据。如下图所示，Kafka 集群中有 4 个 broker，某 topic 有 3 个 partition，且复制因子即副本个数也为 3：</p> <p><img src="https://images.gitbook.cn/616acd70-cf9b-11e8-8388-bd48f25029c6" alt=""></p> <p>如果 leader 所在的 broker 发生故障或宕机，对应 partition 将因无 leader 而不能处理客户端请求，这时副本的作用就体现出来了：一个新 leader 将从 follower 中被选举出来并继续处理客户端的请求。</p> <h4 id="_2-4-数据可靠性保证"><a href="#_2-4-数据可靠性保证" class="header-anchor">#</a> 2.4 数据可靠性保证</h4> <p>一个 partition 有多个副本（replicas），为了提高可靠性，这些副本分散在不同的 broker 上，由于带宽、读写性能、网络延迟等因素，同一时刻，这些副本的状态通常是不一致的：即 followers 与 leader 的状态不一致。</p> <p>为保证 producer 发送的数据，能可靠的发送到指定的 topic，topic 的每个 partition 收到 producer 数据后，都需要向 producer 发送 ack（acknowledgement确认收到），如果 producer 收到 ack，就会进行下一轮的发送，否则重新发送数据。</p> <h5 id="a-副本数据同步策略主要有如下两种"><a href="#a-副本数据同步策略主要有如下两种" class="header-anchor">#</a> a) 副本数据同步策略主要有如下两种</h5> <table><thead><tr><th>方案</th> <th>优点</th> <th>缺点</th></tr></thead> <tbody><tr><td>半数以上完成同步，就发送ack</td> <td>延迟低</td> <td>选举新的 leader 时，容忍n台节点的故障，需要2n+1个副本</td></tr> <tr><td>全部完成同步，才发送ack</td> <td>选举新的 leader 时，容忍n台节点的故障，需要 n+1 个副本</td> <td>延迟高</td></tr></tbody></table> <p>Kafka 选择了第二种方案，原因如下：</p> <ul><li>同样为了容忍 n 台节点的故障，第一种方案需要的副本数相对较多，而 Kafka 的每个分区都有大量的数据，第一种方案会造成大量的数据冗余；</li> <li>虽然第二种方案的网络延迟会比较高，但网络延迟对 Kafka 的影响较小。</li></ul> <h5 id="b-isr"><a href="#b-isr" class="header-anchor">#</a> b) ISR</h5> <p>采用第二种方案之后，设想一下情景：leader 收到数据，所有 follower 都开始同步数据，但有一个 follower 挂了，迟迟不能与 leader 保持同步，那 leader 就要一直等下去，直到它完成同步，才能发送 ack，这个问题怎么解决呢？</p> <p>leader 维护了一个动态的 <strong>in-sync replica set</strong>(ISR)，意为和 leader 保持同步的 follower 集合。当 ISR 中的follower 完成数据的同步之后，leader 就会给 follower 发送 ack。如果 follower 长时间未向 leader 同步数据，则该 follower 将会被踢出 ISR，该时间阈值由 <code>replica.lag.time.max.ms</code> 参数设定。leader 发生故障之后，就会从 ISR 中选举新的 leader。（之前还有另一个参数，0.9 版本之后 <code>replica.lag.max.messages</code> 参数被移除了）</p> <h5 id="c-ack应答机制"><a href="#c-ack应答机制" class="header-anchor">#</a> c) ack应答机制</h5> <p>对于某些不太重要的数据，对数据的可靠性要求不是很高，能够容忍数据的少量丢失，所以没必要等 ISR 中的follower全部接收成功。</p> <p>所以Kafka为用户提供了<strong>三种可靠性级别</strong>，用户根据对可靠性和延迟的要求进行权衡，选择以下的acks 参数配置</p> <ul><li><p>0：producer 不等待 broker 的 ack，这一操作提供了一个最低的延迟，broker 一接收到还没有写入磁盘就已经返回，当 broker 故障时有可能<strong>丢失数据</strong>；</p></li> <li><p>1：producer 等待 broker 的 ack，partition 的 leader 落盘成功后返回 ack，如果在 follower 同步成功之前 leader 故障，那么将会<strong>丢失数据</strong>；</p></li> <li><p>-1（all）：producer 等待 broker 的 ack，partition 的 leader 和 follower 全部落盘成功后才返回 ack。但是	如果在 follower 同步完成后，broker 发送 ack 之前，leader 发生故障，那么就会造成<strong>数据重复</strong>。</p></li></ul> <h5 id="d-故障处理"><a href="#d-故障处理" class="header-anchor">#</a> d) 故障处理</h5> <p>由于我们并不能保证 Kafka 集群中每时每刻 follower 的长度都和 leader 一致（即数据同步是有时延的），那么当leader 挂掉选举某个 follower 为新的 leader 的时候（原先挂掉的 leader 恢复了成为了 follower），可能会出现leader 的数据比 follower 还少的情况。为了解决这种数据量不一致带来的混乱情况，Kafka 提出了以下概念：</p> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh46fmpty5j31eq0hudfw.jpg" alt=""></p> <ul><li>LEO（Log End Offset）：指的是每个副本最后一个offset；</li> <li>HW（High Wather）：指的是消费者能见到的最大的 offset，ISR 队列中最小的 LEO。</li></ul> <p>消费者和 leader 通信时，只能消费 HW 之前的数据，HW 之后的数据对消费者不可见。</p> <p>针对这个规则：</p> <ul><li><strong>当follower发生故障时</strong>：follower 发生故障后会被临时踢出 ISR，待该 follower 恢复后，follower 会读取本地磁盘记录的上次的 HW，并将 log 文件高于 HW 的部分截取掉，从 HW 开始向 leader 进行同步。等该 follower 的 LEO 大于等于该 Partition 的 HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了。</li> <li><strong>当leader发生故障时</strong>：leader 发生故障之后，会从 ISR 中选出一个新的 leader，之后，为保证多个副本之间的数据一致性，其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉，然后从新的 leader 同步数据。</li></ul> <p>所以数据一致性并不能保证数据不丢失或者不重复，这是由 ack 控制的。HW 规则只能保证副本之间的数据一致性！</p> <p>Kafka 的 ISR 的管理最终都会反馈到 ZooKeeper 节点上，具体位置为：</p> <div class="language- extra-class"><pre class="language-text"><code>/brokers/topics/[topic]/partitions/[partition]/state
</code></pre></div><p>目前，有两个地方会对这个 ZooKeeper 的节点进行维护。</p> <ol><li>Controller 来维护：Kafka 集群中的其中一个 Broker 会被选举为 Controller，主要负责 Partition 管理和副本状态管理，也会执行类似于重分配 partition 之类的管理任务。在符合某些特定条件下，Controller 下的 LeaderSelector 会选举新的 leader，ISR 和新的 <code>leader_epoch</code> 及 <code>controller_epoch</code> 写入 ZooKeeper 的相关节点中。同时发起 LeaderAndIsrRequest 通知所有的 replicas。</li> <li>leader 来维护：leader 有单独的线程定期检测 ISR 中 follower 是否脱离 ISR，如果发现 ISR 变化，则会将新的 ISR 的信息返回到 ZooKeeper 的相关节点中。</li></ol> <h4 id="_2-5-exactly-once语义"><a href="#_2-5-exactly-once语义" class="header-anchor">#</a> 2.5 Exactly Once语义</h4> <p>将服务器的 ACK 级别设置为 -1，可以保证 Producer 到 Server 之间不会丢失数据，即 At Least Once 语义。相对的，将服务器 ACK 级别设置为 0，可以保证生产者每条消息只会被发送一次，即 At Most Once语义。</p> <p><strong>At Least Once 可以保证数据不丢失，但是不能保证数据不重复。相对的，At Most Once 可以保证数据不重复，但是不能保证数据不丢失</strong>。但是，对于一些非常重要的信息，比如说交易数据，下游数据消费者要求数据既不重复也不丢失，即 Exactly Once 语义。在 0.11 版本以前的 Kafka，对此是无能为力的，只能保证数据不丢失，再在下游消费者对数据做全局去重。对于多个下游应用的情况，每个都需要单独做全局去重，这就对性能造成了很大的影响。</p> <p>0.11 版本的 Kafka，引入了一项重大特性：<strong>幂等性</strong>。所谓的幂等性就是指 Producer 不论向 Server 发送多少次重复数据。Server 端都会只持久化一条，幂等性结合 At Least Once 语义，就构成了 Kafka 的 Exactily Once 语义，即： <strong><u>At Least Once + 幂等性 = Exactly Once</u></strong></p> <p>要启用幂等性，只需要将 Producer 的参数中 <code>enable.idompotence</code> 设置为 <code>true</code> 即可。Kafka 的幂等性实现其实就是将原来下游需要做的去重放在了数据上游。开启幂等性的 Producer 在初始化的时候会被分配一个 PID，发往同一 Partition 的消息会附带 Sequence Number。而 Broker 端会对 &lt;PID,Partition,SeqNumber&gt; 做缓存，当具有相同主键的消息提交时，Broker 只会持久化一条。</p> <blockquote><p>但是 PID 重启就会变化，同时不同的 Partition 也具有不同主键，所以幂等性无法保证跨分区会话的 Exactly Once。</p></blockquote> <h3 id="三、broker-保存消息"><a href="#三、broker-保存消息" class="header-anchor">#</a> 三、Broker 保存消息</h3> <h4 id="_3-1-存储方式"><a href="#_3-1-存储方式" class="header-anchor">#</a> 3.1 存储方式</h4> <p>物理上把 topic 分成一个或多个 patition（对应 server.properties 中的 num.partitions=3 配置），每个 patition 物理上对应一个文件夹（该文件夹存储该 patition 的所有消息和索引文件）。</p> <h4 id="_3-2-存储策略"><a href="#_3-2-存储策略" class="header-anchor">#</a> 3.2 存储策略</h4> <p>无论消息是否被消费， kafka 都会保留所有消息。有两种策略可以删除旧数据：</p> <ol><li><p>基于时间： <code>log.retention.hours=168</code></p></li> <li><p>基于大小： <code>log.retention.bytes=1073741824</code> 需要注意的是，因为 Kafka 读取特定消息的时间复杂度为 $O(1)$，即与文件大小无关， 所以这里删除过期文件与提高 Kafka 性能无关。</p></li></ol> <h3 id="四、kafka-消费过程"><a href="#四、kafka-消费过程" class="header-anchor">#</a> 四、Kafka 消费过程</h3> <p><strong>Kafka 消费者采用 pull 拉模式从 broker 中消费数据</strong>。与之相对的 push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由 broker 决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成 consumer 来不及处理消息。而 pull 模式则可以根据 consumer 的消费能力以适当的速率消费消息。</p> <p>pull 模式不足之处是，如果 kafka 没有数据，消费者可能会陷入循环中，一直返回空数据。为了避免这种情况，我们在我们的拉请求中有参数，允许消费者请求在等待数据到达的“长轮询”中进行阻塞（并且可选地等待到给定的字节数，以确保大的传输大小，或者传入等待超时时间）。</p> <h4 id="_4-1-消费者组"><a href="#_4-1-消费者组" class="header-anchor">#</a> 4.1 消费者组</h4> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh4aejp3muj31aq0om41k.jpg" alt=""></p> <p>消费者是以 consumer group 消费者组的方式工作，由一个或者多个消费者组成一个组， 共同消费一个 topic。每个分区在同一时间只能由 group 中的一个消费者读取，但是多个 group 可以同时消费这个 partition。在图中，有一个由三个消费者组成的 group，有一个消费者读取主题中的两个分区，另外两个分别读取一个分区。某个消费者读取某个分区，也可以叫做某个消费者是某个分区的拥有者。</p> <p>在这种情况下，消费者可以通过水平扩展的方式同时读取大量的消息。另外，如果一个消费者失败了，那么其他的 group 成员会自动负载均衡读取之前失败的消费者读取的分区。</p> <p><strong>消费者组最为重要的一个功能是实现广播与单播的功能</strong>。一个消费者组可以确保其所订阅的 Topic 的每个分区只能被从属于该消费者组中的唯一一个消费者所消费；如果不同的消费者组订阅了同一个 Topic，那么这些消费者组之间是彼此独立的，不会受到相互的干扰。</p> <blockquote><p>如果我们希望一条消息可以被多个消费者所消费，那么可以将这些消费者放到不同的消费者组中，这实际上就是广播的效果；如果希望一条消息只能被一个消费者所消费，那么可以将这些消费者放到同一个消费者组中，这实际上就是单播的效果。</p></blockquote> <h4 id="_4-2-分区分配策略"><a href="#_4-2-分区分配策略" class="header-anchor">#</a> 4.2 分区分配策略</h4> <p>一个 consumer group 中有多个 consumer，一个 topic 有多个 partition，所以必然会涉及到 partition 的分配问题，即确定哪个 partition 由哪个 consumer 来消费。</p> <p>Kafka 有两种分配策略，一是 RoundRobin，一是 Range（新版本还有Sticky）。</p> <h5 id="roundrobin"><a href="#roundrobin" class="header-anchor">#</a> RoundRobin</h5> <p>RoundRobin 即轮询的意思，比如现在有一个三个消费者 ConsumerA、ConsumerB 和 ConsumerC 组成的消费者组，同时消费 TopicA 主题消息，TopicA 分为 7 个分区，如果采用 RoundRobin 分配策略，过程如下所示：</p> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh47iuetprj31es0ko74s.jpg" alt="图片：mrbird.cc"></p> <p>这种轮询的方式应该很好理解。但如果消费者组消费多个主题的多个分区，会发生什么情况呢？比如现在有一个两个消费者 ConsumerA 和 ConsumerB 组成的消费者组，同时消费 TopicA 和 TopicB 主题消息，如果采用RoundRobin 分配策略，过程如下所示：</p> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh4avsimvoj31ey0ladgc.jpg" alt=""></p> <blockquote><p>注：TAP0 表示 TopicA Partition0 分区数据，以此类推。</p></blockquote> <p>这种情况下，采用 RoundRobin 算法分配，多个主题会被当做一个整体来看，这个整体包含了各自的 Partition，比如在 Kafka-clients 依赖中，与之对应的对象为 <code>TopicPartition</code>。接着将这些 <code>TopicPartition</code> 根据其哈希值进行排序，排序后采用轮询的方式分配给消费者。</p> <p>但这会带来一个问题：假如上图中的消费者组中，ConsumerA 只订阅了 TopicA 主题，ConsumerB 只订阅了TopicB 主题，采用 RoundRobin 轮询算法后，可能会出现 ConsumerA 消费了 TopicB 主题分区里的消息，ConsumerB 消费了 TopicA 主题分区里的消息。</p> <p>综上所述，RoundRobin 算法只适用于消费者组中消费者订阅的主题相同的情况。同时会发现，采用 RoundRobin 算法，消费者组里的消费者之间消费的消息个数最多相差 1 个。</p> <h5 id="range"><a href="#range" class="header-anchor">#</a> Range</h5> <p>Kafka 默认采用 Range 分配策略，Range 顾名思义就是按范围划分的意思。</p> <p>比如现在有一个三个消费者 ConsumerA、ConsumerB 和 ConsumerC 组成的消费者组，同时消费 TopicA 主题消息，TopicA分为7个分区，如果采用 Range 分配策略，过程如下所示：</p> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh47xvny6fj31eo0kot93.jpg" alt=""></p> <p>假如现在有一个两个消费者 ConsumerA 和 ConsumerB 组成的消费者组，同时消费 TopicA 和 TopicB 主题消息，如果采用 Range 分配策略，过程如下所示：</p> <p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh47xrhr61j31fa0lamxi.jpg" alt=""></p> <p>Range 算法并不会把多个主题分区当成一个整体。</p> <p>从上面的例子我们可以总结出Range算法的一个弊端：那就是同一个消费者组内的消费者消费的消息数量相差可能较大。</p> <h4 id="_4-3-offset-的维护"><a href="#_4-3-offset-的维护" class="header-anchor">#</a> 4.3 offset 的维护</h4> <p>由于 consumer 在消费过程中可能会出现断电宕机等故障，consumer 恢复后，需要从故障前的位置继续消费，所以 consumer 需要实时记录自己消费到了哪个 offset，以便故障恢复后继续消费。</p> <p>Kafka 0.9 版本之前，consumer 默认将 offset 保存在 Zookeeper 中，从 0.9 版本开始，consumer 默认将 offset保存在 Kafka 一个内置的 topic 中，该 topic 为 <strong>_consumer_offsets</strong>。</p> <div class="language-shell extra-class"><pre class="language-shell"><code><span class="token operator">&gt;</span> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic starfish --from-beginning
one
</code></pre></div><p>消费 topic 后，查看 kafka-logs 目录，会发现多出 50 个分区。</p> <p>默认情况下__consumer_offsets 有 50 个分区，如果你的系统中 consumer group 也很多的话，那么这个命令的输出结果会很多，具体放置在哪个分区，根据 groupID 做如下计算得到：</p> <div class="language- extra-class"><pre class="language-text"><code>Math.abs(groupID.hashCode()) % numPartitions
</code></pre></div><p><img src="https://tva1.sinaimg.cn/large/007S8ZIlly1gh485ouy5aj31aq0u04qp.jpg" alt=""></p> <h4 id="_4-4-再均衡-rebalance"><a href="#_4-4-再均衡-rebalance" class="header-anchor">#</a> 4.4 再均衡 Rebalance</h4> <p>所谓的再平衡，指的是在 kafka consumer 所订阅的 topic 发生变化时发生的一种分区重分配机制。一般有三种情况会触发再平衡：</p> <ul><li>consumer group 中的新增或删除某个 consumer，导致其所消费的分区需要分配到组内其他的 consumer上；</li> <li>consumer 订阅的 topic 发生变化，比如订阅的 topic 采用的是正则表达式的形式，如 <code>test-*</code> 此时如果有一个新建了一个topic <code>test-user</code>，那么这个 topic 的所有分区也是会自动分配给当前的 consumer 的，此时就会发生再平衡；</li> <li>consumer 所订阅的 topic 发生了新增分区的行为，那么新增的分区就会分配给当前的 consumer，此时就会触发再平衡。</li></ul> <p>Kafka提供的再平衡策略主要有三种：<code>Round Robin</code>，<code>Range</code>和<code>Sticky</code>，默认使用的是<code>Range</code>。这三种分配策略的主要区别在于：</p> <ul><li><code>Round Robin</code>：会采用轮询的方式将当前所有的分区依次分配给所有的consumer；</li> <li><code>Range</code>：首先会计算每个consumer可以消费的分区个数，然后按照顺序将指定个数范围的分区分配给各个consumer；</li> <li><code>Sticky</code>：这种分区策略是最新版本中新增的一种策略，其主要实现了两个目的：
<ul><li>将现有的分区尽可能均衡的分配给各个consumer，存在此目的的原因在于<code>Round Robin</code>和<code>Range</code>分配策略实际上都会导致某几个consumer承载过多的分区，从而导致消费压力不均衡；</li> <li>如果发生再平衡，那么重新分配之后在前一点的基础上会尽力保证当前未宕机的consumer所消费的分区不会被分配给其他的consumer上；</li></ul></li></ul> <h3 id="五、kafka事务"><a href="#五、kafka事务" class="header-anchor">#</a> 五、Kafka事务</h3> <p>Kafka 从 0.11 版本开始引入了事务支持。事务可以保证 Kafka 在 Exactly Once 语义的基础上，生产和消费可以跨分区和会话，要么全部成功，要么全部失败。</p> <h4 id="_5-1-producer事务"><a href="#_5-1-producer事务" class="header-anchor">#</a> 5.1 Producer事务</h4> <p>为了了实现跨分区跨会话的事务，需要引入一个全局唯一的 TransactionID，并将 Producer 获得的 PID 和Transaction ID 绑定。这样当 Producer 重启后就可以通过正在进行的 TransactionID 获得原来的 PID。</p> <p>为了管理 Transaction，Kafka 引入了一个新的组件 Transaction Coordinator。Producer 就是通过和 Transaction Coordinator 交互获得 Transaction ID 对应的任务状态。Transaction Coordinator 还负责将事务所有写入 Kafka 的一个内部 Topic，这样即使整个服务重启，由于事务状态得到保存，进行中的事务状态可以得到恢复，从而继续进行。</p> <h4 id="_5-2-consumer事务"><a href="#_5-2-consumer事务" class="header-anchor">#</a> 5.2 Consumer事务</h4> <p>对 Consumer 而言，事务的保证就会相对较弱，尤其是无法保证 Commit 的消息被准确消费。这是由于Consumer 可以通过 offset 访问任意信息，而且不同的 SegmentFile 生命周期不同，同一事务的消息可能会出现重启后被删除的情况。</p> <p><strong>参考：</strong></p> <p>尚硅谷Kafka教学</p> <p>部分图片来源：mrbird.cc</p> <p>https://gitbook.cn/books/5ae1e77197c22f130e67ec4e/index.html</p></div> <footer class="page-edit" style="display:none;"><!----> <!----></footer> <!----> <!----> <!----></main> <!----></div></div></div></div><div class="global-ui"><div class="back-to-ceiling" style="right:1rem;bottom:6rem;width:2.5rem;height:2.5rem;border-radius:.25rem;line-height:2.5rem;display:none;" data-v-db14854a data-v-db14854a><svg t="1574745035067" viewBox="0 0 1024 1024" version="1.1" xmlns="http://www.w3.org/2000/svg" p-id="5404" class="icon" data-v-db14854a><path d="M526.60727968 10.90185116a27.675 27.675 0 0 0-29.21455937 0c-131.36607665 82.28402758-218.69155461 228.01873535-218.69155402 394.07834331a462.20625001 462.20625001 0 0 0 5.36959153 69.94390903c1.00431239 6.55289093-0.34802892 13.13561351-3.76865779 18.80351572-32.63518765 54.11355614-51.75690182 118.55860487-51.7569018 187.94566865a371.06718723 371.06718723 0 0 0 11.50484808 91.98906777c6.53300375 25.50556257 41.68394495 28.14064038 52.69160883 4.22606766 17.37162448-37.73630017 42.14135425-72.50938081 72.80769204-103.21549295 2.18761121 3.04276886 4.15646224 6.24463696 6.40373557 9.22774369a1871.4375 1871.4375 0 0 0 140.04691725 5.34970492 1866.36093723 1866.36093723 0 0 0 140.04691723-5.34970492c2.24727335-2.98310674 4.21612437-6.18497483 6.3937923-9.2178004 30.66633723 30.70611158 55.4360664 65.4791928 72.80769147 103.21549355 11.00766384 23.91457269 46.15860503 21.27949489 52.69160879-4.22606768a371.15156223 371.15156223 0 0 0 11.514792-91.99901164c0-69.36717486-19.13165746-133.82216804-51.75690182-187.92578088-3.42062944-5.66790279-4.76302748-12.26056868-3.76865837-18.80351632a462.20625001 462.20625001 0 0 0 5.36959269-69.943909c-0.00994388-166.08943902-87.32547796-311.81420293-218.6915546-394.09823051zM605.93803103 357.87693858a93.93749974 93.93749974 0 1 1-187.89594924 6.1e-7 93.93749974 93.93749974 0 0 1 187.89594924-6.1e-7z" p-id="5405" data-v-db14854a></path><path d="M429.50777625 765.63860547C429.50777625 803.39355007 466.44236686 1000.39046097 512.00932183 1000.39046097c45.56695499 0 82.4922232-197.00623328 82.5015456-234.7518555 0-37.75494459-36.9345906-68.35043303-82.4922232-68.34111062-45.57627738-0.00932239-82.52019037 30.59548842-82.51086798 68.34111062z" p-id="5406" data-v-db14854a></path></svg></div><!----></div></div>
    <script src="/assets/js/app.447d4224.js" defer></script><script src="/assets/js/3.9d76740c.js" defer></script><script src="/assets/js/1.c4fd7d2e.js" defer></script><script src="/assets/js/155.c68bf602.js" defer></script>
  </body>
</html>
